package com.behavioranalysis.flinkprogram.util;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.behavioranalysis.flinkprogram.conf.ConfigurationManager;
import com.behavioranalysis.flinkprogram.constant.Constants;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.utils.ParameterTool;

/**
 * 参数工具类
 */
public class ParamUtils {

    /**
     * 从命令行参数中提取任务id
     *
     * @param args     命令行参数
     * @param taskType 任务id
     * @return
     */
    public static Long getTaskIdFromArgs(ExecutionEnvironment env, String[] args, String taskType) {
        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        env.getConfig().setGlobalJobParameters(parameterTool);
        boolean local = ConfigurationManager.getBoolean(Constants.FLINK_LOCAL);
        if (local) {
            return ConfigurationManager.getLong(taskType);
        } else {
            try {
                if (parameterTool.has("input")) {
                    return Long.valueOf(parameterTool.get("input"));
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    /**
     * 从JSON对象中提取参数
     *
     * @param jsonObject JSON对象
     * @param field      参数
     * @return
     */
    public static String getParam(JSONObject jsonObject, String field) {
        JSONArray jsonArray = jsonObject.getJSONArray(field);
        if (jsonArray != null && jsonArray.size() > 0) {
            return jsonArray.getString(0);
        }
        return null;
    }
}
